package com.xxx.datalake.service;

import com.xxx.datalake.entity.Metadata;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.springframework.stereotype.Service;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;

/**
 * @ Project: datalake-unstructured
 * @ Since: 2022/9/13 17:25
 * @ Description:
 */
@Service
public class DataEnterLakeService {

    //用于网页上传，将文件转换为InputStream上传
    //public void updateFile(FileSystem fileSystem, String filename, InputStream in) throws Exception{
    //
    //    //Path pathLocal = new Path("D:\\SHADEILS_pop100_H100_t0.01_F1_1r1.txt");
    //    //InputStream in = new FileInputStream(String.valueOf(pathLocal));
    //    //Path pathServe = new Path("/movie/"+pathLocal.getName());
    //
    //    Path pathServe = new Path("/testFileUpload/"+filename);
    //    FSDataOutputStream out = fileSystem.create(pathServe);
    //    IOUtils.copyBytes(in, out, 2048, true);
    //}

    /**
     * 入湖和提取元数据
     * @param fileSystem    hdfs文件系统
     * @param localPath     本地路径
     * @param hdfsDir       要上传到的hdfs目录
     */
    public Metadata EnterLakeAndStoreMetadata(FileSystem fileSystem, Path localPath, Path hdfsDir) {
        Metadata metadata = new Metadata();
        //提取数据原始名称
        metadata.setOriginalFileName(localPath.getName());
        //提取数据原始位置
        metadata.setOriginalLocation(String.valueOf(localPath.getParent()));
        try {
            //上传
            FileInputStream inputStream = new FileInputStream(String.valueOf(localPath));
            Path hdfsPath = new Path(hdfsDir + "/" + localPath.getName());
            FSDataOutputStream outputStream = fileSystem.create(hdfsPath);
            IOUtils.copyBytes(inputStream,outputStream,2048,true);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return metadata;
    }


}
